package day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object MobileLocation {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("MobileLocation").setMaster("local")
    val sc = new SparkContext(conf)

    //先拿到数据。这里关于Linux与Windows的斜线注意的地方：
    //因为Linux里面只识别/而不识别\，而我们从Windows拷贝的目录过来就是\，（拷贝为字符串变成两个反斜线\\），
    // 所以最好还是在字符串里改成Linux能识别的/
    val dataSet: RDD[String] = sc.textFile("E://大数据培训视频/spark/sparkcoursesinfo/spark/data/lacduration/log")

    //18688888888,20160327082400,16030401EAFB68F1E3CDF819735E1C66,1
    //有了数据，把里面的取出来，变成一个key-value形式，然后根据phone和基站id使用reduceByKey(_+_)得到该phone在该基站待过的总时间
    val splitedUserInfo: RDD[((String, String), Long)] = dataSet.map(x => {
      val lineArr = x.split(",")
      val phone: String = lineArr(0) //phone
      val lac: String = lineArr(2) //基站id
      val flag = lineArr(3).toInt //处于基站里标志
      //时间戳，这里因为要计算在在一个基站待过的时间，那么肯定是为flag为0的时间戳减去flag为1的时间戳得到时间
      //因此在这里为了方便聚合，在这里把这个flag为1的时间戳的值变成负数，然后让聚合的时候让相加就得到待过的总时间。
      val timeStamp: Long = if (flag == 1) -(lineArr(1).toLong) else lineArr(1).toLong
      //返回组成的(基站id,phone,时间戳)，因为要进行join，所以必须是二元组。所以key为(phone,lac)，value为timeStamp
      ((phone, lac), timeStamp)
    })

    //计算该phone在该基站待过的总时间，然后重新整理数据要join的格式
    val aggreData: RDD[((String, String), Long)] = splitedUserInfo.reduceByKey(_+_)
    val lacAndPhoneTime: RDD[(String, (String, Long))] = aggreData.map(x => {
      val phone: String = x._1._1
      val lac: String = x._1._2
      val time: Long = x._2
      (lac, (phone, time))
    })

    //我们要用join算子来将另外一个含有基站经纬度数据与本数据按照基站id去join到一起
    //得到基站经纬度数据，并map切分，返回二元组(基站id,(经度,纬度)),因为使用join算子数据类型必须是k-v形式，
    //9F36407EAD8829FC166F14DDE7970F68,116.304864,40.050645,6
    val lacInfo = sc.textFile("E://大数据培训视频/spark/sparkcoursesinfo/spark/data/lacduration/lac_info.txt")
    val splitedLacInfoData = lacInfo.map(x => {
      val lineArr = x.split(",")
      val lac: String = lineArr(0) //基站id
      val X: String = lineArr(1) //经度
      val Y: String = lineArr(2) //纬度
      //返回二元组，因为要join
      (lac, (X, Y))
    })
    //join合并，(k,v)与(k,m)进行join操作后返回值是(k,(k,v),(k,m))
    val joinedData: RDD[(String, ((String, Long), (String, String)))] = lacAndPhoneTime.join(splitedLacInfoData)
    //把这个数据整合一下变成(phone,lac,time,x,y)
    val lacPoneTimeXY: RDD[(String, String, Long, String, String)] = joinedData.map(x => {
      val lac: String = x._1
      val phone: String = x._2._1._1
      val time: Long = x._2._1._2
      val X: String = x._2._2._1
      val Y: String = x._2._2._2
      (phone, lac, time, X, Y)
    })

    //为了统计用户在所有基站的信息，所以要生成用户粒度的信息(phone,[phone,lac,time,X,Y])
    val UserGrouped: RDD[(String, Iterable[(String, String, Long, String, String)])] = lacPoneTimeXY.groupBy(_._1)

    //按照时长进行组内排序，注意是组内组内组内，重要的事情说三遍。（也就是说计算每一个人按照时长排序）
    //即phone --> 按照time排序
    //所以要用mapValues而不能直接用sortBy
    //注意Iterable变成List才能用sort排序
    val sortedData: RDD[(String, List[(String, String, Long, String, String)])] = UserGrouped.mapValues(_.toList.sortBy(_._3).reverse)

    //获得top2即每个phone的前两个待的时间最长的信息。注意打印的话要将Array转成toBuffer
    val res: mutable.Buffer[(String, List[(String, String, Long, String, String)])] = sortedData.mapValues(_.take(2)).collect().toBuffer
    res.foreach(println)
    sc.stop()
  }
}
